package com.tangding.service.db;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

import org.bson.Document;
import org.bson.conversions.Bson;

import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.WriteConcern;
import com.mongodb.client.AggregateIterable;
import com.mongodb.client.DistinctIterable;
import com.mongodb.client.FindIterable;
import com.mongodb.client.ListCollectionsIterable;
import com.mongodb.client.ListIndexesIterable;
import com.mongodb.client.MapReduceIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.IndexOptions;
import com.mongodb.client.model.Projections;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.geojson.Position;
import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.UpdateResult;
import com.tangding.common.utils.JacksonUtil;

/**
 * @description 
 *
 * @author Ranger
 * @version 2017年8月18日_下午6:22:20
 *
 */
public class MongoDBUtil {
	
	/**
	 * 今天做MongoDB并发测试，报出上述错误。究其原因，是数据库连接数太少，资源耗尽。查看com.mongodb.MongoOptions源代码，
	 * 其中有connectionsPerHost和threadsAllowedToBlockForConnectionMultiplier两个重要的属性。
	 * connectionsPerHost：每个主机的连接数
	 * threadsAllowedToBlockForConnectionMultiplier ：线程队列数，它以上面connectionsPerHost值相乘的结果就是线程队列最大值。如果连接线程排满了队列就会抛出“Out of semaphores to get db”错误。 connectionsPerHost默认是10，
	 * threadsAllowedToBlockForConnectionMultiplier 默认是5，也就是线程池有50个连接数可供使用。因此只要将这个属性的值加大就可以避免上述错误。
	 * 
	 * 其它属性设置： 
	 * maxWaitTime:最大等待连接的线程阻塞时间
	 * connectTimeout：连接超时的毫秒。0是默认和无限
	 * socketTimeout：socket超时。0是默认和无限 
	 * autoConnectRetry：这个控制是否在一个连接时，系统会自动重试
	 * 
	 * */
	public static MongoClient createMongoClientDetails(String host, int port) {
		MongoClientOptions.Builder build = MongoClientOptions.builder();
		build.connectionsPerHost(100);   //与目标数据库能够建立的最大connection数量为50
//      build.autoConnectRetry(true);   //自动重连数据库启动  
        build.threadsAllowedToBlockForConnectionMultiplier(50); //如果当前所有的connection都在使用中，则每个connection上可以有50个线程排队等待  
		/*
		* 一个线程访问数据库的时候，在成功获取到一个可用数据库连接之前的最长等待时间为2分钟 
		* 这里比较危险，如果超过maxWaitTime都没有获取到这个连接的话，该线程就会抛出Exception 
		* 故这里设置的maxWaitTime应该足够大，以免由于排队线程过多造成的数据库访问失败 
		*/
        build.maxWaitTime(1000 * 60 * 2);  
        build.connectTimeout(1000 * 60 * 1);    //与数据库建立连接的timeout设置为1分钟  
        build.writeConcern(WriteConcern.ACKNOWLEDGED);
        
        MongoClientOptions options = build.build();
        
        //下面的server可以是一个list，主要用于mongos集群
        ServerAddress server = new ServerAddress(host, port);
        
        /** 连接池默认10 */
		MongoClient mongo = new MongoClient(server, options);
		return mongo;
	}
	
	public static MongoClient createMongoClient(String host) {
        MongoClientOptions.Builder build = MongoClientOptions.builder();
        build.writeConcern(WriteConcern.ACKNOWLEDGED);
        MongoClient client = new MongoClient(host, build.build());
		return client;
	}
	
	public static MongoClient createMongoClient(String host, int port) {
        ServerAddress server = new ServerAddress(host, port);
        MongoClientOptions.Builder build = MongoClientOptions.builder();
        build.writeConcern(WriteConcern.ACKNOWLEDGED);
        MongoClient client = new MongoClient(server, build.build());
		return client;
	}
	
	public static MongoClient createMongosClient(List<ServerAddress> addresses) {
        MongoClientOptions.Builder build = MongoClientOptions.builder();
        build.writeConcern(WriteConcern.ACKNOWLEDGED);
        MongoClient client = new MongoClient(addresses, build.build());
        return client;
	}
	
	public static MongoClient createMongosClient(List<ServerAddress> addresses, List<MongoCredential> credentialsList) {
		MongoClientOptions.Builder build = MongoClientOptions.builder();
        build.writeConcern(WriteConcern.ACKNOWLEDGED);
        MongoClient client = new MongoClient(addresses, credentialsList, build.build());
        return client;
	}
	
	/**
	 * 获取指定DB下所有的collection
	 * 
	 * @param client
	 * @param option
	 * @return
	 */
	public static ListCollectionsIterable<Document> listCollections(MongoClient client, String DB) {
		ListCollectionsIterable<Document> documents = client.getDatabase(DB).listCollections();
		return documents;
	}
	
	/**
	 * 插入
	 * 
	 * @param client
	 * @param option
	 * @param data
	 * @param id
	 */
	public static void insert(MongoClient client, MongoDBOption option, Document data) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		documents.insertOne(data);
	}
	
	/**
	 * 批量插入
	 * 
	 * @param client
	 * @param option
	 * @param datas
	 */
	public static void insertMany(MongoClient client, MongoDBOption option, List<Document> datas) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		documents.insertMany(datas);
	}
	
	
	/**
	 * 删除
	 * 
	 * @param client
	 * @param option
	 * @param condition
	 * @return
	 */
	public static <T extends Bson> DeleteResult remove(MongoClient client, MongoDBOption option, T condition) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return documents.deleteOne(condition);
	}
	
	/**
	 * 删除全部记录
	 * 
	 * @param mc
	 * @param option
	 * @return
	 */
	public static <T extends Bson> DeleteResult removeMany(MongoClient mc, MongoDBOption option, T condition) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.deleteMany(condition);
	}
	
	/**
	 * 删除全部记录
	 * 
	 * @param mc
	 * @param option
	 * @return
	 */
	public static DeleteResult removeAll(MongoClient mc, MongoDBOption option) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.deleteMany(new Document());
	}
	
	/**
	 * 删除collection
	 * 
	 * @param mc
	 * @param option
	 */
	public static void drop(MongoClient mc, MongoDBOption option) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		collection.drop();
	}
	
	/**
	 * 替换
	 * 
	 * @param mc
	 * @param option
	 * @param data
	 * @param condition
	 * @return
	 */
	public static <T extends Bson> UpdateResult replace(MongoClient mc, MongoDBOption option, Document data, T condition) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.replaceOne(condition, data);
	}
	
	public static <T extends Bson> UpdateResult replace(MongoClient mc, MongoDBOption option, Document data, T condition, UpdateOptions updateOption) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.replaceOne(condition, data, updateOption);
	}
	
	/**
	 * 更新
	 * 
	 * @param client
	 * @param option
	 * @param data
	 * @param id
	 * @return
	 */
	public static <T extends Bson> UpdateResult update(MongoClient client, MongoDBOption option, Document data, T condition) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return documents.updateOne(condition, new Document("$set", data), new UpdateOptions().upsert(true));
	}

	/**
	 * 更新
	 * 
	 * @param client
	 * @param option
	 * @param data
	 * @param id
	 * @return
	 */
	public static <T extends Bson> UpdateResult update(MongoClient client, MongoDBOption option, Document data, T condition, UpdateOptions updateOptions) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return documents.updateOne(condition, new Document("$set", data), updateOptions);
	}
	
	/**
	 * 更新多条。很多情况下用updateMany可以代替updateOne
	 * 
	 * @param client
	 * @param option
	 * @param data
	 * @param id
	 * @return
	 */
	public static <T extends Bson> UpdateResult updateMany(MongoClient client, MongoDBOption option, Document data, T condition) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return documents.updateMany(condition, new Document("$set", data), new UpdateOptions().upsert(true));
	}
	
	/**
	 * 查找
	 * 
	 * @param client
	 * @param option
	 * @param id
	 * @return
	 */
	public static <T extends Bson> FindIterable<Document> find(MongoClient client, MongoDBOption option, T condition) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return documents.find(condition);
	}
	
	/**
	 * 查找并且返回指定的属性值
	 * 
	 * @param client
	 * @param option
	 * @param id
	 * @return
	 */
	public static <T extends Bson> FindIterable<Document> find(MongoClient client, MongoDBOption option, T condition, String... files) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return documents.find(condition).projection(Projections.fields(Projections.include(files)));
	}
	
	/**
	 * 指定记录总条数
	 * @param client
	 * @param dbName
	 * @param collectionName
	 * @param condition
	 * @return
	 */
	public static <T extends Bson> long count(MongoClient client, MongoDBOption option, T condition) {
		MongoCollection<Document> collection = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.count(condition);
	}
	
	/**
	 * 分页查询
	 * 
	 * @param client
	 * @param option
	 * @param condition
	 * @param skip
	 * @param limit
	 * @param files
	 * @return
	 */
	public static <T extends Bson> FindIterable<Document> find(MongoClient client, MongoDBOption option, T condition, int skip, int limit, String... files) {
		MongoCollection<Document> documents = client.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return documents.find(condition).projection(Projections.fields(Projections.include(files))).skip(skip).limit(limit);
	}
	
	/**
	 * 分页查询
	 * 
	 * @param client
	 * @param dbName
	 * @param collectionName
	 * @param condition
	 * @return
	 */
	public static <T extends Bson> AggregateIterable<Document> page(MongoClient client, MongoDBOption option, int start, int pageSize, T condition) {
		List<Bson> params = new ArrayList<Bson>();
		if (condition != null) {
			Aggregates.match(condition);
			params.add(Aggregates.match(condition));
		}
		params.add(Aggregates.skip(start));
		params.add(Aggregates.limit(pageSize));
		return aggregate(client, option, params);
	}
	
	/**
	 * 去重返回指定字段列表
	 * 
	 * @param mc
	 * @param dbName
	 * @param collectionName
	 * @param filed
	 * @param type
	 * @return
	 */
	public static <T> DistinctIterable<T> distinct(MongoClient mc, MongoDBOption option, String filed, Class<T> type) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.distinct(filed, type);
	}
	
	/**
	 * mapreduce操作具体可以参照有道云笔记
	 * 
	 * @param mc
	 * @param dbName
	 * @param collectionName
	 * @param mkey
	 * @param mvalue
	 * @param rkey
	 * @param rvalue
	 * @return
	 */
	public static MapReduceIterable<Document> mapreduce(MongoClient mc, MongoDBOption option, String mkey, String mvalue, String rkey, String rvalue) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		String map = "function() {  emit(this." + mkey + ", this." + mvalue + ");}";
		String reduce = "function(key, values) { return {" + rkey + ":key," + rvalue + ":values};} ";  
		return collection.mapReduce(map, reduce);
	}
	
	/**
	 * 接收可变性更高的mapreduce参数
	 * 
	 * @param mc
	 * @param dbName
	 * @param collectionName
	 * @param map
	 * @param reduce
	 * @return
	 */
	public static MapReduceIterable<Document> mapreduceX(MongoClient mc, MongoDBOption option, String map, String reduce) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.mapReduce(map, reduce);
	}
	
	
	/**
	 * aggregate通道
	 * 
	 * @param mc
	 * @param option
	 * @param params
	 * @return
	 */
	public static AggregateIterable<Document> aggregate(MongoClient mc, MongoDBOption option, List<? extends Bson> params) {
		MongoCollection<Document> collection = mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName());
		return collection.aggregate(params);
	}
	
	/**
	 * 列出索引
	 * 
	 * @param mc
	 * @param option
	 * @return
	 */
	public static ListIndexesIterable<Document> listIndexes(MongoClient mc, MongoDBOption option) {
		return mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName()).listIndexes();
	}
	
	
	/**
	 * 创建索引
	 * 
	 * @param mc
	 * @param option
	 * @param keys
	 * @return 返回indexName,如：创建索引{"name":1}，indexName=name_1
	 */
	public static <T extends Bson> String createIndex(MongoClient mc, MongoDBOption option, T keys) {
		return mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName()).createIndex(keys);
	}
	
	/**
	 * 创建索引
	 * 
	 * @param mc
	 * @param option
	 * @param keys
	 * @param indexOptions
	 * @return 返回indexName,如：创建索引{"name":1}，indexName=name_1
	 */
	public static <T extends Bson> String createIndex(MongoClient mc, MongoDBOption option, T keys, IndexOptions indexOptions) {
		return mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName()).createIndex(keys, indexOptions);
	}
	
	/**
	 * 删除索引
	 * 
	 * @param mc
	 * @param option
	 * @param indexName
	 */
	public static void dropIndex(MongoClient mc, MongoDBOption option, String indexName) {
		mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName()).dropIndex(indexName);
	}
	
	/**
     * 查询在圆形区域内的坐标点，需要指定中心点坐标和半径，半径单位是米（此方法球半径为地球半径）
     *
     * @param locationField 坐标字段
     * @param center        中心点坐标[经度，纬度]
     * @param radius        半径 单位:米
     * @param fields        查询字段
     * @param condition     查询条件
     * @param limit         返回记录限制数量
     * @return              非NULL的list
     */
    public static FindIterable<Document> withinCircle(MongoClient mc, MongoDBOption option, String locationField, Point center, long radius, Document fields, Document condition) {
        return withinCircle(mc, option, locationField, center, radius, 6378137.0f, fields, condition, 0, 10);
    }
    
    
    /**
     * 查询在圆形区域内的坐标点，需要指定中心点坐标和半径，球的半径
     * 
     * @param locationField 坐标字段
     * @param center		中心点坐标[经度，纬度]
     * @param radius		半径 单位
     * @param ballRadius	球的半径
     * @param fields		查询字段
     * @param condition		查询条件
     * @param skip			跳过查询
     * @param limit			限定查询
     * @return
     */
    public static FindIterable<Document> withinCircle(MongoClient mc, MongoDBOption option, String locationField, Point center, long radius, float ballRadius, Document fields, Document condition, int skip, int limit) {
    	LinkedList<Object> circle = new LinkedList<Object>();
        //Set the center coordinate
        circle.addLast(center.getPointList());
        //Set the radius. unit:meter
        circle.addLast(radius/ballRadius);
 
		if (condition == null) {
			condition = new Document();
		}
        condition.put(locationField, new Document("$geoWithin", new Document("$centerSphere", circle)));
        return mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName()).find(condition).projection(fields).skip(skip).limit(limit);
    }
    
    /**
	 * 查询点是否在多边形内
	 * 
	 * @param mc
	 * 			连接客户端
	 * @param option
	 * 			查询的库和集合对象
	 * @param locationField
	 *            坐标字段
	 * @param lng 经度
	 * @param lat 纬度           
	 * @return
	 */
	public static FindIterable<Document> withinPolygon(MongoClient mc, MongoDBOption option, String locationField, double lng, double lat) {
		BasicDBList point = new BasicDBList();
		point.add(lng);
		point.add(lat);
        BasicDBObject query = new BasicDBObject();
		query.put(locationField, new BasicDBObject("$geoIntersects", new BasicDBObject("$geometry", new BasicDBObject("type","Point").append("coordinates",point))));
		
		/*List<Double> values = new ArrayList<Double>();
		values.add(lng);
		values.add(lat);
		List<Position> positions = new ArrayList<Position>();
		positions.add(new Position(values));
		@SuppressWarnings("unchecked")
		PolygonCoordinates coordinates = new PolygonCoordinates(positions);
		Bson query = Filters.geoIntersects(locationField, new Polygon(coordinates));*/
		return mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName()).find(query).projection(Projections.fields(Projections.exclude("_id")));
	}
	
	
	/**
	 * 查询环形内的点
	 * 
	 * @param distance 单位 : 米
	 */
	public static FindIterable<Document> withinAnnular(MongoClient mc, MongoDBOption option, String locationField, double lng, double lat, double maxDistance, double minDistance) {
//		BasicDBList point = new BasicDBList();
//		point.add(lng);
//		point.add(lat);
//		BasicDBObject query = new BasicDBObject();
//		query.put(locationField, new BasicDBObject("$near", new BasicDBObject("$geometry", new BasicDBObject("type","Point").append("coordinates",point)).append("$minDistance", minDistance).append("$maxDistance", maxDistance)));
		List<Double> values = new ArrayList<>();
		values.add(lng);
		values.add(lat);
		Bson query = Filters.near(locationField, new com.mongodb.client.model.geojson.Point(new Position(values)), maxDistance, minDistance);
		return mc.getDatabase(option.getDbName()).getCollection(option.getCollectionName()).find(query).projection(Projections.fields(Projections.exclude("_id")));
	}
    
    public static class Point {
    	private double longitude;//经度
    	private double latitude;//纬度
    	
    	public Point(double longitude, double latitude) {
    		super();
    		this.longitude = longitude;
    		this.latitude = latitude;
    	}

    	public LinkedList<Double> getPointList() {
    		LinkedList<Double> p = new LinkedList<Double>();
    		p.addLast(longitude);
    		p.addLast(latitude);
    		return p;
    	}
    	
    	/**
         * 转换为GeoJSON对象
    	 * @throws Exception 
         */
        public Document getGeoJson(){
        	Document loc=new Document();
    		loc.append("type", "Point");
    		loc.append("coordinates", getPointList());
    		return loc;
        }
    }
    
    public static void main(String[] args) {
		MongoClient client = createMongoClient("172.16.189.201", 17272);
		double longitude = 113.338954;
		double latitude = 23.143425;
		MongoDBOption option = new MongoDBOption("share", "operation");
		FindIterable<Document> list = withinCircle(client, option, "location", new Point(longitude, latitude), 5000, new Document(), new Document());
		System.out.println(JacksonUtil.toJson(list));
	}
}